package service;
import org.apache.kafka.clients.consumer.*;
import java.sql.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class KafkaToMySQLConsumer {
    private static final String JDBC_URL = "jdbc:mysql://192.168.244.11:3306/stats?serverTimezone=Asia/Shanghai";
    private static final String USER = "root";
    private static final String PASSWORD = "123456";

    public static void main(String[] args) {
        // Kafka消费者配置
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.244.11:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "mysql-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringDeserializer");

        // MySQL连接
        Connection connection = null;
        try {
            connection = DriverManager.getConnection(JDBC_URL, USER, PASSWORD);
            // 关闭自动提交
            connection.setAutoCommit(false);

            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

            // 准备SQL语句
            String sql = "INSERT INTO student_info(class_id, name, gender, birth_date, " +
                    "student_id, semester, gpa, enrollment_status) VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
            PreparedStatement statement = connection.prepareStatement(sql);

            // 订阅主题
            consumer.subscribe(Collections.singletonList("t2025"));

            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    String[] data = record.value().split("\t");

                    // 设置参数
                    statement.setString(1, data[0]); // class_id
                    statement.setString(2, data[1]); // name
                    statement.setInt(3, Integer.parseInt(data[2])); // gender
                    statement.setDate(4, Date.valueOf(data[3])); // birth_date
                    statement.setString(5, data[4]); // student_id
                    statement.setInt(6, Integer.parseInt(data[5])); // semester
                    statement.setDouble(7, Double.parseDouble(data[6])); // gpa
                    statement.setString(8, data[7]); // enrollment_status

                    // 执行插入
                    statement.executeUpdate();
                    System.out.println("Inserted record: " + record.value());
                }
                // 手动提交事务
                connection.commit();
            }
        } catch (Exception e) {
            e.printStackTrace();
            // 发生异常时回滚
            if (connection != null) {
                try {
                    connection.rollback();
                } catch (SQLException ex) {
                    ex.printStackTrace();
                }
            }
        } finally {
            // 关闭连接
            if (connection != null) {
                try {
                    connection.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}